package com.dc.di.sf.dt.convertor;

import java.util.ArrayList;
import java.util.List;

import org.pentaho.di.core.BlockingRowSet;
import org.pentaho.di.core.RowMetaAndData;
import org.pentaho.di.core.RowSet;
import org.pentaho.di.core.row.RowMeta;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMeta;
import org.pentaho.di.core.row.ValueMetaInterface;

import com.dc.di.sf.dt.BusinessData;
import com.dc.di.sf.dt.DataSet;
import com.dc.di.sf.dt.DataStructure;
import com.dc.di.sf.dt.DataUnit;
import com.dc.di.sf.dt.RecordData;
import com.dc.di.sf.dt.UnitData;

public class TransObjectConvertor {

	/**
	 * @param transResult
	 * @return
	 */
	public static BusinessData convertTransResultToBusinessData(List<RowMetaAndData> transResult) {
		if (transResult == null || transResult.isEmpty())
			return null;
		BusinessData businessData = new BusinessData();
		DataStructure dataStructure = convertRowMetaToDataStructure(transResult.get(0).getRowMeta());
		businessData.setDataStructure(dataStructure);
		businessData.setDataSet(new DataSet(dataStructure));
		for (RowMetaAndData rowMetaAndData : transResult) {
			RecordData recordData = convertRowDataToRecordData(dataStructure, rowMetaAndData.getData());
			businessData.getDataSet().addRecordData(recordData);
		}
		return businessData;
	}

	/**
	 * @param rowMeta
	 * @return
	 */
	public static DataStructure convertRowMetaToDataStructure(RowMetaInterface rowMeta) {
		if (rowMeta == null)
			return null;
		DataStructure dataStructure = new DataStructure();
		for (ValueMetaInterface valueMeta : rowMeta.getValueMetaList())
			dataStructure.addDataUnit(convertValueMetaToDataUnit(valueMeta));
		return dataStructure;
	}

	public static RecordData convertRowDataToRecordData(DataStructure dataStructure, Object[] rowData) {
		if (rowData == null)
			return null;
		RecordData recordData = new RecordData(dataStructure);
		List<DataUnit> dataUnits = dataStructure.getDataUnits();
		for (int i = 0; i < dataUnits.size(); i++) {
			UnitData unitData = new UnitData(dataUnits.get(i));
			unitData.setUnitValue(rowData[i]);
			recordData.addUnitData(unitData);
		}
		return recordData;
	}

	/**
	 * @param businessData
	 * @return
	 */
	public static List<RowMetaAndData> convertBusinessDataToTransResult(BusinessData businessData) {
		if (businessData == null)
			return null;
		List<RowMetaAndData> transResult = new ArrayList<RowMetaAndData>();
		RowMetaInterface rowMeta = convertDataStructureToRowMeta(businessData.getDataStructure());
		for (RecordData recordData : businessData.getDataSet().getRecordDatas()) {
			Object[] rowData = convertRecordDataToRowData(rowMeta, recordData);
			transResult.add(new RowMetaAndData(rowMeta, rowData));
		}
		return transResult;
	}

	public static RowMetaInterface convertDataStructureToRowMeta(DataStructure dataStructure) {
		if (dataStructure == null)
			return null;
		RowMetaInterface rowMeta = new RowMeta();
		for (DataUnit dataUnit : dataStructure.getDataUnits())
			rowMeta.addValueMeta(convertDataUnitToValueMeta(dataUnit));
		return rowMeta;
	}

	public static Object[] convertRecordDataToRowData(RowMetaInterface rowMeta, RecordData recordData) {
		if (recordData == null)
			return null;
		List<UnitData> unitDatas = recordData.getUnitDatas();
		if (unitDatas == null)
			return null;
		Object[] rowData = new Object[unitDatas.size()];
		for (UnitData unitData : unitDatas) {
			int i = rowMeta.indexOfValue(unitData.getUnitIdName());
			if (i >= 0)
				rowData[i] = unitData.getUnitValue();
		}
		return rowData;
	}

	public static DataSet convertRowSetToDataSet(RowSet rowSet) {
		DataStructure dataStructure = convertRowMetaToDataStructure(rowSet.getRowMeta());
		DataSet dataSet = new DataSet(dataStructure);

		Object[] rowData = null;
		do {
			rowData = getRowFrom(rowSet);
			if (rowData != null)
				dataSet.addRecordData(convertRowDataToRecordData(dataStructure, rowData));
		} while (rowData != null);
		return dataSet;
	}

	public static RowSet convertDataSetToRowSet(RowMetaInterface rowMeta, DataSet dataSet) {
		int size = dataSet.getRecordDatas().size();
		RowSet rowSet = new BlockingRowSet(size <=0 ? 1 : size);
		rowSet.setRowMeta(rowMeta);
		for (RecordData recordData : dataSet.getRecordDatas()) {
			rowSet.putRow(rowMeta, convertRecordDataToRowData(rowMeta, recordData));
		}
		return rowSet;
	}

	public static DataUnit convertValueMetaToDataUnit(ValueMetaInterface valueMeta) {
		DataUnit dataUnit = new DataUnit();
		dataUnit.setIdName(valueMeta.getName());
		switch (valueMeta.getType()) {
		case ValueMetaInterface.TYPE_STRING:
			dataUnit.setDataType(DataUnit.TYPE_STRING);
			break;
		case ValueMetaInterface.TYPE_BIGNUMBER:
			dataUnit.setDataType(DataUnit.TYPE_DECIMAL);
			break;
		case ValueMetaInterface.TYPE_DATE:
			dataUnit.setDataType(DataUnit.TYPE_DATETIME);
			break;
		case ValueMetaInterface.TYPE_BINARY:
			dataUnit.setDataType(DataUnit.TYPE_BINARY);
			break;
		//add by liyang
		case ValueMetaInterface.TYPE_INTEGER:
			dataUnit.setDataType(DataUnit.TYPE_INTEGER);
			break;
		case ValueMetaInterface.TYPE_NUMBER:
			dataUnit.setDataType(DataUnit.TYPE_NUMBER);
			break;
		}
		return dataUnit;
	}

	public static ValueMetaInterface convertDataUnitToValueMeta(DataUnit unit) {
		ValueMetaInterface valueMeta = new ValueMeta();
		valueMeta.setName(unit.getIdName());
		String type = unit.getDataType();
		if (DataUnit.TYPE_STRING.equals(type))
			valueMeta.setType(ValueMetaInterface.TYPE_STRING);
		else if (DataUnit.TYPE_DECIMAL.equals(type))
			valueMeta.setType(ValueMetaInterface.TYPE_BIGNUMBER);
		else if (DataUnit.TYPE_DATETIME.equals(type))
			valueMeta.setType(ValueMetaInterface.TYPE_DATE);
		else if (DataUnit.TYPE_BINARY.equals(type))
			valueMeta.setType(ValueMetaInterface.TYPE_BINARY);
		//add by liyang
		else if (DataUnit.TYPE_INTEGER.equals(type))
			valueMeta.setType(ValueMetaInterface.TYPE_INTEGER);
		else if (DataUnit.TYPE_NUMBER.equals(type))
			valueMeta.setType(ValueMetaInterface.TYPE_NUMBER);
		else
			valueMeta.setType(ValueMetaInterface.TYPE_NONE);

		return valueMeta;
	}

	private static Object[] getRowFrom(RowSet rowSet) {

		Object[] rowData = rowSet.getRow();
//		while (rowData == null && !rowSet.isDone()) {
//			rowData = rowSet.getRow();
//		}
//		if (rowData == null && rowSet.isDone()) {
//			// Try one more time to get a row to make sure we don't get a
//			// race-condition between the get and the isDone()
//			//
//			rowData = rowSet.getRow();
//		}
//		if (rowData == null && rowSet.isDone()) {
//			// Try one more time...
//			rowData = rowSet.getRow();
//			if (rowData == null) {
//				return null;
//			}
//		}
		return rowData;
	}

}
